package cn.doitedu.demo.groupby_agg;

import org.apache.flink.streaming.api.CheckpointingMode;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.table.api.bridge.java.StreamTableEnvironment;

public class 滚动时间窗口 {
    public static void main(String[] args) {

        // 构造stream api 编程的环境
        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
        env.enableCheckpointing(5000, CheckpointingMode.EXACTLY_ONCE);
        env.getCheckpointConfig().setCheckpointStorage("file:///d:/ckpt");
        env.setParallelism(1);

        // 构造 sql 编程环境
        StreamTableEnvironment tenv = StreamTableEnvironment.create(env);
        // 针对kafka source这种类型的需要读取多个“分区”数据的source，
        // 当有一个分区超过 1ms没有数据，则这个分区的event_time不参与watermark的选择
        tenv.getConfig().set("table.exec.source.idle-timeout","1 ms");


        // 创建数据源映射表
        tenv.executeSql("CREATE TABLE t1                  (\n" +
                "  uid BIGINT,                               \n" +
                "  event_id STRING,                          \n" +
                "  properties MAP<STRING,STRING>,            \n" +
                "  action_time BIGINT,                       \n" +

                "  rt as to_timestamp_ltz(action_time,3),    \n" +
                "  watermark for rt as rt                    \n" +
                ") WITH (                                    \n" +
                "  'connector' = 'kafka',                    \n" +
                "  'topic' = 'ss-1',                         \n" +
                "  'properties.bootstrap.servers' = 'doitedu:9092',\n" +
                "  'properties.group.id' = 'doit44_g1',      \n" +
                "  'scan.startup.mode' = 'latest-offset',    \n" +
                "  'value.format' = 'json',                  \n" +
                "  'value.fields-include' = 'EXCEPT_KEY'     \n" +
                ")");


        // 执行查询sql
        tenv.executeSql(
                     " SELECT                                                "+
                        "   window_start,                                       "+
                        " 	window_end,                                         "+
                        " 	count(distinct uid) as uv,                          "+
                      //" 	count(if(event_id='page_load',1,null)) as pv        "+ // 报错：不能直接使用null
                      //" 	sum(if(event_id='page_load',1,0)) as pv             "+  // 可以
                        " 	count(1) filter(where event_id='page_load') as pv   "+  // 可以
                      //" 	sum(1) filter(where event_id='page_load') as pv     "+  // 可以
                        " FROM TABLE(                                           "+
                        "   TUMBLE(TABLE t1,DESCRIPTOR(rt),INTERVAL '1' MINUTE) "+
                        " )                                                     "+
                        " GROUP BY                                              "+
                        "     window_start,                                     "+
                        " 	  window_end                                        "
        );
    }
}
